package com.hsogoo.avatar.registry.zookeeper;

import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;

import com.hsogoo.avatar.model.AvatarAddress;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;

import com.hsogoo.avatar.exceptions.AvatarException;
import com.hsogoo.avatar.registry.AbstractAvatarRegistry;
import com.hsogoo.avatar.registry.RegisterMeta;
import com.hsogoo.avatar.registry.ServerListChangeListener;
import com.hsogoo.avatar.registry.ServiceMeta;

/**
 * @Author huangsaigang
 * Created @2018/9/25.
 */
public class ZookeeperAvatarRegistry extends AbstractAvatarRegistry {

    private ZkClient zkClient;

    @Override
    public void connectToRegistryServer(String connectString) {
        zkClient = ZkClient.getInstance(connectString);
    }

    @Override
    protected void doRegister(RegisterMeta meta) {
        if(zkClient == null){
            throw new AvatarException(String.format("havn't connect to registry server ..."));
        }
        ServiceMeta serviceMeta = meta.getServiceMeta();
        String directory = String.format("/avatar/provider/%s/%s/%s",
                serviceMeta.getGroup(),
                serviceMeta.getServiceProviderName(),
                serviceMeta.getVersion());
        String node = String.format("%s/%s:%s",directory, meta.getAddress().getHost(), meta.getAddress().getPort());
        zkClient.createNode(node, CreateMode.EPHEMERAL, new BackgroundCallback(){

            @Override
            public void processResult(CuratorFramework curatorFramework, CuratorEvent event) throws Exception {
                if (event.getResultCode() == KeeperException.Code.OK.intValue()) {
                    //设置状态为完成
                    getRegisterMetaMap().put(meta, RegisterState.DONE);
                }
            }
        });
    }

    @Override
    protected void doCheckRegisterNodeStatus() {
        for (Map.Entry<RegisterMeta, RegisterState> entry : getRegisterMetaMap().entrySet()) {
            if (entry.getValue() == RegisterState.DONE) {
                continue;
            }

            RegisterMeta meta = entry.getKey();
            String directory = String.format("/avatar/provider/%s/%s/%s",
                    meta.getServiceMeta().getGroup(),
                    meta.getServiceMeta().getServiceProviderName(),
                    meta.getServiceMeta().getVersion());

            String nodePath = String.format("%s/%s:%s",
                    directory,
                    meta.getAddress().getHost(),
                    String.valueOf(meta.getAddress().getPort()));
            if(zkClient.checkNodeExist(nodePath)){
                entry.setValue(RegisterState.DONE);
            }else{
                //重新注册服务
                register(meta);
            }
        }
    }

    @Override
    protected void doSubscribe(ServiceMeta serviceMeta) {
        String directory = String.format("/avatar/provider/%s/%s/%s",
                serviceMeta.getGroup(),
                serviceMeta.getServiceProviderName(),
                serviceMeta.getVersion());
        zkClient.watchServerList(directory, new ServerListChangeListener(){

            @Override
            public void onChange(ServerChangeEvent event, String data) {
                RegisterMeta registerMeta = parseRegisterMeta(data);
                switch (event){
                    case SERVER_ADD:
                        providerAdded(registerMeta);
                        break;
                    case SERVER_REMOVED:
                        providerRemoved(registerMeta);
                        break;
                }
            }
        });
    }

    private RegisterMeta parseRegisterMeta(String data) {
        String[] array_0 = data.split("/");
        RegisterMeta registerMeta = new RegisterMeta();
        ServiceMeta serviceMeta = new ServiceMeta();
        serviceMeta.setGroup(array_0[3]);
        serviceMeta.setServiceProviderName(array_0[4]);
        serviceMeta.setVersion(array_0[5]);

        String[] array_1 = array_0[6].split(":");
        AvatarAddress address = new AvatarAddress();
        address.setHost(array_1[0]);
        address.setPort(Integer.parseInt(array_1[1]));

        registerMeta.setAddress(address);
        registerMeta.setServiceMeta(serviceMeta);

        return registerMeta;
    }

}
